package com.nl.pulsardemo;

import com.nl.pulsardemo.pool.config.produce.MyProducerPoolConfig;
import com.nl.pulsardemo.pool.config.produce.PooledProducerFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.pulsar.client.api.*;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
@Slf4j
class PulsarDemoApplicationTest2 {
    @Autowired
    private PulsarClient client;

    @Test
    void contextLoads() throws Exception {
//        GenericObjectPool<Producer> op = new GenericObjectPool<>(new PooledProducerFactory(), new GenericObjectPoolConfig());
        GenericObjectPool<Producer> op = new GenericObjectPool<>(new PooledProducerFactory(), new MyProducerPoolConfig());
        // 从连接池获取对象
        Producer producer1 = op.borrowObject();
        Producer producer2 = op.borrowObject();
        System.out.println("producer1:" + producer1.getProducerName());
//         归还给连接池
        op.returnObject(producer1);
        Producer producer3 = op.borrowObject();
        System.out.println("producer2:" + producer2.getProducerName());
        System.out.println("producer3:" + producer3.getProducerName());

//        for (int i = 0; i < 8; i++) {
//            Producer producer = op.borrowObject();
//
//            System.out.println("producer" + i + ":" + producer.getProducerName());
//        }
        SyncSend(producer1);
    }

    /**
     * @param
     * @detail 同步发送
     * @author ZhaoY
     * @Return
     */

    public void SyncSend(Producer producer) throws PulsarClientException {
        for (int i = 1; i < 30; i++) {
            String content = "" + i;
            producer.send(content.getBytes());//同步发送
            log.info("Send message: " + content);
        }
        producer.close();//关闭producer
    }
}
